style(#2591): reword render_dag_line doc so clippy doesn't read it as a list

Two tests asserted the old behaviour where completed nodes/DAGs stayed
in the snapshot. Under the redesign, Done nodes are filtered off the wire
(a fully-Done DAG disappears; a Failed one lingers + is history-capped):
- failed_node test: the completed reconcile is Done → assert it's absent,
  not Done-present (its run is already verified by the claim).
- history-eviction test: fail the nodes so the DAGs linger (Done ones
  would vanish), then assert the grace window + per-template cap.

hive-c0re/src/dashboard/build_logs.rs

@@ -90,6 +90,42 @@ pub(super) async fn get_build_log_full(
     }
 }
 
+/// `GET /api/build-log/{node_id}` — the build log for a **queue node**,
+/// resolved node id → log-row id → full log. Same `BuildLogFull` JSON
+/// (`stdout` / `stderr` + header) as `get_build_log_full`; HTTP 404 when the
+/// node has no linked log (the client gates the request on `NodeView.has_log`,
+/// but a vacuum race can still 404). This is the on-demand log fetch the
+/// raw-graph dashboard uses instead of an inline `build_log_id` on the wire.
+pub(super) async fn get_build_log_for_node(
+    State(state): State<AppState>,
+    AxumPath(node_id): AxumPath<u64>,
+) -> Response {
+    match state.coord.job_queue.build_log_id_of(node_id) {
+        Some(log_id) => get_build_log_full(State(state), AxumPath(log_id)).await,
+        None => (
+            StatusCode::NOT_FOUND,
+            format!("node #{node_id} has no build log"),
+        )
+            .into_response(),
+    }
+}
+
+/// `GET /api/build-log/{node_id}/raw` — the node's build log as `text/plain`
+/// for download (delegates to `get_build_log_raw` after resolving the node id).
+pub(super) async fn get_build_log_raw_for_node(
+    State(state): State<AppState>,
+    AxumPath(node_id): AxumPath<u64>,
+) -> Response {
+    match state.coord.job_queue.build_log_id_of(node_id) {
+        Some(log_id) => get_build_log_raw(State(state), AxumPath(log_id)).await,
+        None => (
+            StatusCode::NOT_FOUND,
+            format!("node #{node_id} has no build log"),
+        )
+            .into_response(),
+    }
+}
+
 /// JSON frame sent on the `/api/build-logs/id/{id}/stream` SSE channel.
 /// `stdout_append` / `stderr_append` carry only the new bytes since the
 /// last frame; `done = true` means the build finished and the stream

hive-c0re/src/dashboard/mod.rs

@@ -100,6 +100,14 @@ pub async fn serve(
         )
         .route("/api/audit-log", get(misc_api::api_audit_log))
         .route("/api/build-logs", get(build_logs::get_build_logs_all))
+        .route(
+            "/api/build-log/{node_id}",
+            get(build_logs::get_build_log_for_node),
+        )
+        .route(
+            "/api/build-log/{node_id}/raw",
+            get(build_logs::get_build_log_raw_for_node),
+        )
         .route(
             "/api/build-logs/{agent}",
             get(build_logs::get_build_logs_agent),

hive-c0re/src/job_queue/mod.rs

@@ -38,6 +38,7 @@ mod tests;
 use std::collections::HashMap;
 use std::sync::Mutex;
 
+use chrono::{DateTime, Utc};
 use hive_jobq::resources::ResourceTable;
 use hive_jobq::scheduler::{Outcome, Scheduler};
 use hive_jobq::{Dep, DepWhen as JobDepWhen, Graph, NodeId, State as JobState};
@@ -96,15 +97,14 @@ pub struct TerminalDag {
     pub error: Option<String>,
 }
 
-/// Per-node runtime metadata the crate graph doesn't carry (kind + agent live
-/// in the node payload; state lives in the node).
+/// Per-node runtime metadata the crate graph doesn't carry. Lifecycle
+/// (`started_at` / `finished_at` / `error`) lives on the `hive_jobq::Node`
+/// itself now, so only the two host-side extras remain: the live sub-step
+/// label and the build-log row link (the client fetches the log by node id).
 #[derive(Debug, Default, Clone)]
 struct NodeRuntime {
     step: Option<String>,
     build_log_id: Option<i64>,
-    started_at: Option<i64>,
-    finished_at: Option<i64>,
-    error: Option<String>,
 }
 
 /// An owned read-view of a DAG container's carried metadata ([`NodeKind::Dag`]).
@@ -362,7 +362,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 {
@@ -386,9 +385,8 @@ impl JobQueue {
                 inputs: meta.inputs,
                 transient: meta.transient,
             });
-            if let Some(rt) = inner.node_rt.get_mut(&id) {
-                rt.started_at = Some(now);
-            }
+            // `started_at` is stamped on the graph `Node` by the scheduler's
+            // transition to `Running` — no host-side copy needed.
         }
         claims
     }
@@ -405,23 +403,15 @@ impl JobQueue {
         result: Result<(), String>,
     ) -> Option<TerminalDag> {
         let mut inner = self.lock();
-        let now = now_unix();
-        let (error, outcome) = match result {
-            Ok(()) => (None, Outcome::Done),
-            Err(e) => {
-                // The reason rides the crate `Outcome::Failed` (stamped onto the
-                // graph `Node`); the `node_rt` copy stays for now until the wire
-                // reads it off the node directly.
-                let msg = truncate_error(&e);
-                (Some(msg.clone()), Outcome::Failed(msg))
-            }
+        // The failure reason + `finished_at` are stamped onto the graph `Node`
+        // by the scheduler (the reason rides `Outcome::Failed`); no host-side
+        // copy. We only clear the live sub-step label here.
+        let outcome = match result {
+            Ok(()) => Outcome::Done,
+            Err(e) => Outcome::Failed(truncate_error(&e)),
         };
         if let Some(rt) = inner.node_rt.get_mut(&node_id) {
-            rt.finished_at = Some(now);
             rt.step = None;
-            if let Some(e) = error {
-                rt.error = Some(e);
-            }
         }
         let container = inner.dag_of(node_id);
         inner.sched.complete(node_id, outcome);
@@ -521,6 +511,21 @@ impl JobQueue {
         true
     }
 
+    /// The `build_logs` row id linked to the wire node id `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). Takes the raw wire `u64` (the endpoint's path
+    /// param); `node_rt` is keyed by the opaque `NodeId`, so this scans for the
+    /// matching id — the map is small (live + recently-terminal nodes).
+    #[must_use]
+    pub fn build_log_id_of(&self, node_id: u64) -> Option<i64> {
+        self.lock()
+            .node_rt
+            .iter()
+            .find(|(nid, _)| nid.get() == node_id)
+            .and_then(|(_, rt)| rt.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.
@@ -721,15 +726,13 @@ impl QueueInner {
         seen
     }
 
-    /// First failed work node's stored error, for the roll-up `error` field.
+    /// First failed work node's error (read off the graph `Node`), for the
+    /// terminal roll-up summary the inline hook consumes.
     fn dag_first_error(&self, container: NodeId) -> Option<String> {
         for id in self.subtree(container) {
-            if self
-                .sched
-                .graph()
-                .node(id)
-                .is_some_and(|n| n.state == JobState::Failed)
-                && let Some(e) = self.node_rt.get(&id).and_then(|r| r.error.clone())
+            if let Some(n) = self.sched.graph().node(id)
+                && n.state == JobState::Failed
+                && let Some(e) = n.error.clone()
             {
                 return Some(e);
             }
@@ -749,19 +752,35 @@ impl QueueInner {
         })
     }
 
-    /// Rebuild the wire [`DagView`] for a DAG from its container metadata + work
-    /// nodes + per-node runtime.
+    /// Project a DAG into its wire [`DagView`]: a near-raw view of the
+    /// container's work nodes, with `Done` nodes excluded. Lifecycle
+    /// (`state` / `started_at` / `finished_at` / `error`) is read straight
+    /// off each `hive_jobq::Node`; the client derives the DAG label, roll-up
+    /// state, and DAG timestamps from the node set. Non-derivable per-node
+    /// payload (`approval_id`, meta `inputs`) rides the owning node. Returns
+    /// `None` when every work node is `Done` — a fully-completed DAG drops
+    /// out of the snapshot entirely (a `Failed` one lingers until aged out).
     fn dag_view(&self, container: NodeId) -> Option<DagView> {
         let meta = self.dag_meta(container)?;
-        let node_ids = self.subtree(container);
-        let mut nodes = Vec::with_capacity(node_ids.len());
-        let mut started: Vec<i64> = Vec::new();
-        let mut finished: Vec<i64> = Vec::new();
-        for &id in &node_ids {
+        let mut nodes = Vec::new();
+        // DAG-level timestamps are taken over *all* subtree nodes (including the
+        // `Done` ones excluded from the wire) — the client can't derive them
+        // from a `Done`-filtered node set, so the host computes them here.
+        let mut started: Vec<DateTime<Utc>> = Vec::new();
+        let mut finished: Vec<DateTime<Utc>> = Vec::new();
+        for id in self.subtree(container) {
             let Some(node) = self.sched.graph().node(id) else {
                 continue;
             };
-            let rt = self.node_rt.get(&id);
+            if let Some(s) = node.started_at {
+                started.push(s);
+            }
+            if let Some(f) = node.finished_at {
+                finished.push(f);
+            }
+            if node.state == JobState::Done {
+                continue;
+            }
             let deps: Vec<u64> = node
                 .deps
                 .iter()
@@ -770,51 +789,54 @@ impl QueueInner {
                     Dep::Resource { .. } => None,
                 })
                 .collect();
-            if let Some(s) = rt.and_then(|r| r.started_at) {
-                started.push(s);
-            }
-            if let Some(fin) = rt.and_then(|r| r.finished_at) {
-                finished.push(fin);
-            }
+            // Non-derivable per-node payload rides the node that owns it.
+            let approval_id = matches!(node.payload, NodeKind::ApprovalDeploy { .. })
+                .then_some(meta.approval_id)
+                .flatten();
+            let inputs = if matches!(node.payload, NodeKind::MetaLock { .. }) {
+                meta.inputs.clone()
+            } else {
+                Vec::new()
+            };
+            let has_log = self.node_rt.get(&id).and_then(|r| r.build_log_id).is_some();
             nodes.push(NodeView {
                 id: id.get(),
                 agent: node.payload.agent().to_owned(),
                 kind: node.payload.as_str().to_owned(),
                 deps,
                 state: to_wire_state(node.state),
-                step: rt.and_then(|r| r.step.clone()),
-                build_log_id: rt.and_then(|r| r.build_log_id),
-                started_at: rt.and_then(|r| r.started_at),
-                finished_at: rt.and_then(|r| r.finished_at),
-                error: rt.and_then(|r| r.error.clone()),
+                started_at: node.started_at,
+                finished_at: node.finished_at,
+                error: node.error.clone(),
+                approval_id,
+                inputs,
+                has_log,
             });
         }
+        if nodes.is_empty() {
+            return None;
+        }
         let is_terminal = self.dag_is_terminal(container);
         Some(DagView {
             id: container.get(),
-            kind: meta.template,
-            state: self.dag_rollup(container),
             source: meta.source,
             reason: meta.reason.clone(),
-            enqueued_at: meta.created_at,
+            created_at: hive_sh4re::wire_time::from_secs(meta.created_at),
             started_at: started.into_iter().min(),
-            finished_at: if is_terminal {
-                finished.into_iter().max()
-            } else {
-                None
-            },
-            inputs: meta.inputs.clone(),
-            approval_id: meta.approval_id,
+            finished_at: is_terminal.then(|| finished.into_iter().max()).flatten(),
             nodes,
         })
     }
 
     /// When a DAG's work node finishes on `finished_at` — the max over its
-    /// subtree, for the history cap ordering.
+    /// subtree (read off the graph `Node`, as unix seconds), for the history
+    /// cap ordering.
     fn dag_finished_at(&self, container: NodeId) -> i64 {
         self.subtree(container)
             .iter()
-            .filter_map(|id| self.node_rt.get(id).and_then(|r| r.finished_at))
+            .filter_map(|id| self.sched.graph().node(*id))
+            .filter_map(|n| n.finished_at)
+            .map(|t| t.timestamp())
             .max()
             .unwrap_or(0)
     }

hive-c0re/src/job_queue/model.rs

@@ -11,11 +11,62 @@
 //! DAG can span agents). See `docs/coordinator.md::Job queue` for the
 //! full design.
 
-pub use hive_sh4re::jobs::{DagView, NodeId, PermPayload, Source, State, Template};
+pub use hive_sh4re::jobs::{DagView, NodeId, PermPayload, Source, State};
 use serde::Serialize;
 
 use crate::coordinator::TransientKind;
 
+/// What a DAG *means* — the request-level shape. Internal to the queue now:
+/// it drives the terminal-hook dispatch ([`crate::job_queue`]'s `dag_hook`)
+/// and the meta-update dedup key, and is **no longer sent on the wire** — the
+/// dashboard derives a DAG's label from its node kinds (see `DagView`). The
+/// `NodeKind::Dag` container carries it in its payload.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize)]
+#[serde(rename_all = "snake_case")]
+pub enum Template {
+    /// Rebuild one agent's container (prebuild → stop → profile-swap →
+    /// reconcile).
+    Rebuild,
+    /// Bump meta flake locks; grows a rebuild subgraph per affected
+    /// agent into the same DAG on completion.
+    MetaUpdate,
+    /// First-deploy spawn (approval-driven).
+    Spawn,
+    /// Mechanical stop + converge to `wanted = Up` (a restart).
+    Restart,
+    /// Signal → drain → mechanical stop → converge to `wanted = Up` — a
+    /// graceful restart as one atomic DAG.
+    GracefulRestart,
+    /// Perm-file commit followed by the rebuild subgraph.
+    PermChange,
+    /// Quiesce the harness, drain, then stop (`wanted = Offline`).
+    GracefulStop,
+    /// Converge to `wanted = Up`.
+    Start,
+    /// Converge to `wanted = Offline`.
+    Stop,
+    /// Boot-time config sweep as one DAG.
+    Boot,
+}
+
+impl Template {
+    #[must_use]
+    pub fn as_str(self) -> &'static str {
+        match self {
+            Template::Rebuild => "rebuild",
+            Template::MetaUpdate => "meta_update",
+            Template::Spawn => "spawn",
+            Template::Restart => "restart",
+            Template::GracefulRestart => "graceful_restart",
+            Template::PermChange => "perm_change",
+            Template::GracefulStop => "graceful_stop",
+            Template::Start => "start",
+            Template::Stop => "stop",
+            Template::Boot => "boot",
+        }
+    }
+}
+
 /// When a dependency edge is considered satisfied.
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
 #[serde(rename_all = "snake_case")]

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")
-        .state
+        .map_or(State::Done, DagView::rollup_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()
@@ -676,7 +679,12 @@ fn failed_node_cancels_downstream_but_afterany_reconcile_runs() {
     assert_eq!(by_kind("stop_for_update"), State::Cancelled);
     assert_eq!(by_kind("swap"), State::Cancelled);
     assert_eq!(by_kind("post_swap"), State::Cancelled);
-    assert_eq!(by_kind("reconcile"), State::Done);
+    // The AfterAny reconcile ran (claimed + completed Ok above) → it's `Done`,
+    // and `Done` nodes are excluded from the wire, so it's absent here.
+    assert!(
+        dag.nodes.iter().all(|n| n.kind != "reconcile"),
+        "the completed (Done) reconcile is filtered off the wire"
+    );
     assert_eq!(
         dag.nodes
             .iter()
@@ -872,10 +880,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 +898,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();
-    let node = &snap.iter().find(|d| d.id == id).expect("dag").nodes[0];
-    assert_eq!(node.build_log_id, Some(43), "log id survives completion");
+    // The log id is fetched by node id (the `GET /api/build-log/<id>` lookup),
+    // not carried on the wire — it survives completion in the node runtime.
+    assert_eq!(
+        q.build_log_id_of(c.node_id.get()),
+        Some(43),
+        "log id survives completion"
+    );
 }
 
 #[test]
@@ -910,9 +922,12 @@ fn history_evicts_old_terminals_per_template() {
             ),
         );
         let c = claim_one(&q);
-        // Completing the single work node rolls the container up terminal (its
-        // inline hook fires off the returned summary — no terminal-hook node).
-        q.complete_node(id, c.node_id, Ok(()));
+        // Fail the single work node so the DAG *lingers*: a fully-`Done` DAG
+        // drops off the wire entirely, but a `Failed` one is retained (+
+        // history-capped) so the operator can still triage it. Completing the
+        // node rolls the container up terminal (its inline hook fires off the
+        // returned summary — no terminal-hook node).
+        q.complete_node(id, c.node_id, Err("boom".to_owned()));
     }
     // Fresh terminals are inside the grace window: nothing evicts yet,
     // so a ~1s QueueDag poller can still observe every terminal state

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
-            .iter()
-            .any(|id| snap.iter().any(|d| d.id == *id && !d.state.is_terminal()));
+        // A DAG has settled when it's either gone from the snapshot (fully
+        // `Done` DAGs drop out) or still present but with every node 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

@@ -6,68 +6,9 @@
 //! live in `hive-c0re::job_queue`; these are the serialized views it
 //! produces. Semantics: `docs/coordinator.md::Job queue`.
 
+use chrono::{DateTime, Utc};
 use serde::{Deserialize, Serialize};
 
-/// What a DAG *means* — the request-level shape. Wire strings match
-/// the pre-DAG queue's `kind` values so dashboards key off the same
-/// tags.
-#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
-#[serde(rename_all = "snake_case")]
-pub enum Template {
-    /// Rebuild one agent's container (prebuild → stop → profile-swap →
-    /// reconcile).
-    Rebuild,
-    /// Bump meta flake locks; grows a rebuild subgraph per affected
-    /// agent into the same DAG on completion.
-    MetaUpdate,
-    /// First-deploy spawn (approval-driven).
-    Spawn,
-    /// Reserved for a future destroy integration.
-    Destroy,
-    /// Mechanical stop + converge to `wanted = Up` (a restart).
-    Restart,
-    /// Signal → drain → mechanical stop → converge to `wanted = Up` — a
-    /// graceful restart as one atomic DAG (drains the harness before the
-    /// stop, same as `GracefulStop`, but then reconciles back up instead
-    /// of staying down).
-    GracefulRestart,
-    /// Perm-file commit followed by the rebuild subgraph.
-    PermChange,
-    /// Quiesce the harness, drain, then stop (`wanted = Offline`).
-    GracefulStop,
-    /// Converge to `wanted = Up`.
-    Start,
-    /// Converge to `wanted = Offline`.
-    Stop,
-    /// Bare converge of observed power state to the persisted intent
-    /// (boot reconcile).
-    Reconcile,
-    /// Boot-time config sweep as one DAG: a hyperhive lock bump that grows
-    /// a rebuild subgraph per stale agent, plus a `Reconcile` per drifted
-    /// agent — all in a single DAG (no anchor node, no child DAGs).
-    Boot,
-}
-
-impl Template {
-    #[must_use]
-    pub fn as_str(self) -> &'static str {
-        match self {
-            Template::Rebuild => "rebuild",
-            Template::MetaUpdate => "meta_update",
-            Template::Spawn => "spawn",
-            Template::Destroy => "destroy",
-            Template::Restart => "restart",
-            Template::GracefulRestart => "graceful_restart",
-            Template::PermChange => "perm_change",
-            Template::GracefulStop => "graceful_stop",
-            Template::Start => "start",
-            Template::Stop => "stop",
-            Template::Reconcile => "reconcile",
-            Template::Boot => "boot",
-        }
-    }
-}
-
 /// Where the submit request originated — drives the "why" chip on the
 /// dashboard.
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
@@ -140,9 +81,12 @@ pub enum PermPayload {
 /// so the widening from the old dag-local `u32` is transparent.
 pub type NodeId = u64;
 
-/// One node of a queued DAG, as serialized. Step labels, build-log
-/// links, errors, and timestamps are per-node; the DAG-level `state`
-/// is a roll-up.
+/// One node of a queued DAG, serialized near-raw from the scheduler
+/// graph. Lifecycle (`state` / `started_at` / `finished_at` / `error`)
+/// comes straight off the `hive_jobq::Node`. The client derives DAG-level
+/// roll-ups (label, state, timestamps) from the node set — nothing is
+/// rolled up host-side. Build logs are fetched on demand by node id
+/// (`GET /api/build-log/<id>`), not carried inline.
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct NodeView {
     pub id: NodeId,
@@ -156,44 +100,124 @@ pub struct NodeView {
     /// `"drain"`, `"write_dropin"`, `"write_perm_file"`,
     /// `"approval_deploy"`.
     pub kind: String,
-    /// Ids of the nodes this one waits for.
+    /// Ids of the nodes this one waits for. May reference an already-`Done`
+    /// node that's been filtered out of the wire — the client treats a dep
+    /// on an absent node as satisfied.
     #[serde(default)]
     pub deps: Vec<NodeId>,
     pub state: State,
     #[serde(default, skip_serializing_if = "Option::is_none")]
-    pub step: Option<String>,
+    pub started_at: Option<DateTime<Utc>>,
     #[serde(default, skip_serializing_if = "Option::is_none")]
-    pub build_log_id: Option<i64>,
-    #[serde(default, skip_serializing_if = "Option::is_none")]
-    pub started_at: Option<i64>,
-    #[serde(default, skip_serializing_if = "Option::is_none")]
-    pub finished_at: Option<i64>,
+    pub finished_at: Option<DateTime<Utc>>,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub error: Option<String>,
+    /// Approval-queue row id — present only on the `approval_deploy` node.
+    /// The client links a DAG to its pending approval through this (it is
+    /// not derivable from the graph, so it rides the node that owns it).
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub approval_id: Option<i64>,
+    /// Meta-flake inputs being bumped — present only on the `meta_lock`
+    /// node. Display-only payload, not derivable from the graph.
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub inputs: Vec<String>,
+    /// Whether this node has a captured build log fetchable at
+    /// `GET /api/build-log/<id>`. Only the nix-heavy nodes that stream build
+    /// output set one; the client gates its log link on this so lock / noop /
+    /// store-only nodes don't render a link that 404s.
+    #[serde(default)]
+    pub has_log: bool,
 }
 
-/// A queued/running/recent DAG. `kind` = template string, roll-up
-/// `state`; everything per-node appears exactly once, inside `nodes`.
-/// There is no DAG-level `agent` — a DAG can span agents, so agent lives
-/// on each [`NodeView`]; consumers group nodes by `NodeView::agent`.
+/// A queued / running / failed DAG — a thin projection of one container
+/// node plus its (non-`Done`) subtree from the scheduler graph. Only
+/// non-derivable facts live here: `id`, `source`, `reason`, `created_at`,
+/// and the node set. The client derives the card label, roll-up state, and
+/// DAG timestamps from `nodes` (per-node `kind` + lifecycle) — nothing is
+/// rolled up host-side. There is no DAG-level `agent`: a DAG can span
+/// agents, so agent is per-[`NodeView`]; consumers group by `NodeView::agent`.
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct DagView {
     pub id: u64,
-    /// Template wire string — same values the old `kind` field used.
-    pub kind: Template,
-    /// Roll-up: `failed` if any node failed, else `running` /
-    /// `queued` / `cancelled` / `done`.
-    pub state: State,
     pub source: Source,
     pub reason: String,
-    pub enqueued_at: i64,
+    /// When the DAG was enqueued.
+    pub created_at: DateTime<Utc>,
+    /// When the DAG's first node started (min over *all* its nodes) — computed
+    /// host-side, **not** derived on the client: `Done` nodes are excluded from
+    /// `nodes` below, so the earliest-started node is usually absent from the
+    /// wire and the client can't take the min itself. `None` until a node runs.
     #[serde(default, skip_serializing_if = "Option::is_none")]
-    pub started_at: Option<i64>,
+    pub started_at: Option<DateTime<Utc>>,
+    /// When the DAG finished (max `finished_at` over all its nodes), set only
+    /// once the DAG has settled terminal. Host-computed for the same reason as
+    /// `started_at`. `None` while the DAG is still live.
     #[serde(default, skip_serializing_if = "Option::is_none")]
-    pub finished_at: Option<i64>,
-    #[serde(default, skip_serializing_if = "Vec::is_empty")]
-    pub inputs: Vec<String>,
-    #[serde(default, skip_serializing_if = "Option::is_none")]
-    pub approval_id: Option<i64>,
+    pub finished_at: Option<DateTime<Utc>>,
+    /// Nodes of this DAG with `Done` ones excluded. A DAG whose nodes are
+    /// all `Done` is omitted from the snapshot entirely; a `Failed` DAG
+    /// 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
+        }
+    }
+
+    /// A short human label for the DAG, derived from its node kinds — the
+    /// shared replacement for the old `Template` wire string now that the kind
+    /// isn't sent. Priority-ordered so the most distinctive node wins (an
+    /// approval deploy reads as "deploy" even though it also rebuilds). Purely
+    /// cosmetic (progress/queue display); consumers that need exactness inspect
+    /// the node kinds directly.
+    #[must_use]
+    pub fn label(&self) -> &'static str {
+        let has = |k: &str| self.nodes.iter().any(|n| n.kind == k);
+        if has("approval_deploy") {
+            "deploy"
+        } else if has("meta_lock") {
+            "meta-update"
+        } else if has("create") || has("provision") {
+            "spawn"
+        } else if has("swap") || has("prebuild") || has("post_swap") {
+            "rebuild"
+        } else if has("write_perm_file") {
+            "perm-change"
+        } else if has("signal") || has("drain") {
+            "graceful"
+        } else if has("set_wanted") || has("stop_for_update") {
+            "power"
+        } else {
+            "reconcile"
+        }
+    }
+}

hivectl/src/dag_progress.rs

@@ -66,8 +66,8 @@ async fn wait_for_dags_plain(socket: &Path, ids: Vec<u64>) -> Result<()> {
                 // may still be running — keep watching so the operator
                 // sees whether the agent came back.
                 if d.nodes.iter().all(|n| n.state.is_terminal()) {
-                    if d.state == hive_sh4re::jobs::State::Failed {
-                        failed.push(format!("{} {}", d.kind.as_str(), dag_agents(d)));
+                    if d.rollup_state() == hive_sh4re::jobs::State::Failed {
+                        failed.push(format!("{} {}", d.label(), dag_agents(d)));
                     }
                 } else {
                     all_terminal = false;
@@ -134,8 +134,8 @@ async fn wait_for_dags_animated(socket: &Path, ids: Vec<u64>) -> Result<()> {
                 });
                 hdr.set_message(format!(
                     "{} {} {} · {}",
-                    state_glyph(d.state),
-                    d.kind.as_str(),
+                    state_glyph(d.rollup_state()),
+                    d.label(),
                     dag_agents(d),
                     fmt_dur(dag_elapsed(d, now)),
                 ));
@@ -163,8 +163,8 @@ async fn wait_for_dags_animated(socket: &Path, ids: Vec<u64>) -> Result<()> {
                     }
                 }
                 if d.nodes.iter().all(|n| n.state.is_terminal()) {
-                    if d.state == hive_sh4re::jobs::State::Failed {
-                        failed.push(format!("{} {}", d.kind.as_str(), dag_agents(d)));
+                    if d.rollup_state() == hive_sh4re::jobs::State::Failed {
+                        failed.push(format!("{} {}", d.label(), dag_agents(d)));
                     }
                 } else {
                     all_terminal = false;
@@ -224,18 +224,21 @@ fn now_unix() -> i64 {
         .unwrap_or(0)
 }
 
-/// Elapsed seconds for a DAG: `started_at` (falling back to `enqueued_at`)
-/// through `finished_at` or `now`.
+/// Elapsed seconds for a DAG: `started_at` (falling back to `created_at`)
+/// through `finished_at` or `now`. The wire carries these as RFC3339
+/// `DateTime<Utc>`; compare in unix seconds against `now`.
 fn dag_elapsed(d: &hive_sh4re::jobs::DagView, now: i64) -> i64 {
-    let start = d.started_at.unwrap_or(d.enqueued_at);
-    (d.finished_at.unwrap_or(now) - start).max(0)
+    let start = d
+        .started_at
+        .map_or_else(|| d.created_at.timestamp(), |t| t.timestamp());
+    (d.finished_at.map_or(now, |t| t.timestamp()) - start).max(0)
 }
 
 /// Elapsed seconds for a node: `started_at` → `finished_at`/`now`, or 0
 /// when it hasn't started.
 fn node_elapsed(n: &hive_sh4re::jobs::NodeView, now: i64) -> i64 {
     match n.started_at {
-        Some(start) => (n.finished_at.unwrap_or(now) - start).max(0),
+        Some(start) => (n.finished_at.map_or(now, |t| t.timestamp()) - start.timestamp()).max(0),
         None => 0,
     }
 }
@@ -250,16 +253,11 @@ fn fmt_dur(secs: i64) -> String {
     }
 }
 
-/// One animated node line: kind, live step, an `(after …)` marker for a
-/// fan-in node (>1 dep), its elapsed timer, and a truncated error tail.
+/// One animated node line: kind, an `(after …)` marker for a fan-in node
+/// (>1 dep), its elapsed timer, and a truncated error tail.
 fn node_line(d: &hive_sh4re::jobs::DagView, n: &hive_sh4re::jobs::NodeView, now: i64) -> String {
     use std::fmt::Write as _;
     let mut s = n.kind.clone();
-    if n.state == hive_sh4re::jobs::State::Running
-        && let Some(step) = &n.step
-    {
-        let _ = write!(s, " ({step})");
-    }
     if n.deps.len() > 1 {
         let after: Vec<&str> = n
             .deps
@@ -292,25 +290,21 @@ fn state_glyph(state: hive_sh4re::jobs::State) -> &'static str {
     }
 }
 
-/// One progress line for a DAG: roll-up glyph, template, agent, then
-/// the node chain with the running node's live step label — the CLI
-/// twin of the dashboard's queue card. Used by the plain (non-TTY) path.
+/// One progress line for a DAG: roll-up glyph, derived label, agents, then
+/// the node chain — the CLI twin of the dashboard's queue card. Both the
+/// roll-up state and the label are derived from the node set (the wire no
+/// longer carries them). Used by the plain (non-TTY) path.
 fn render_dag_line(d: &hive_sh4re::jobs::DagView) -> String {
     use std::fmt::Write as _;
     let mut out = format!(
         "{} {} {:<12}",
-        state_glyph(d.state),
-        d.kind.as_str(),
+        state_glyph(d.rollup_state()),
+        d.label(),
         dag_agents(d)
     );
     for (i, n) in d.nodes.iter().enumerate() {
         let sep = if i == 0 { "  " } else { " → " };
         let _ = write!(out, "{sep}{} {}", state_glyph(n.state), n.kind);
-        if n.state == hive_sh4re::jobs::State::Running
-            && let Some(step) = &n.step
-        {
-            let _ = write!(out, " ({step})");
-        }
     }
     if let Some(err) = d.nodes.iter().find_map(|n| n.error.as_deref()) {
         let short: String = err.chars().take(120).collect();
@@ -321,76 +315,65 @@ fn render_dag_line(d: &hive_sh4re::jobs::DagView) -> String {
 
 #[cfg(test)]
 mod tests {
-    use hive_sh4re::jobs::{DagView, NodeView, Source, State, Template};
+    use hive_sh4re::jobs::{DagView, NodeView, Source, State};
+    use hive_sh4re::wire_time::from_secs;
 
     use super::render_dag_line;
 
-    fn node(id: u64, agent: &str, kind: &str, state: State, step: Option<&str>) -> NodeView {
+    fn node(id: u64, agent: &str, kind: &str, state: State) -> NodeView {
         NodeView {
             id,
             agent: agent.to_owned(),
             kind: kind.to_owned(),
             deps: if id == 0 { vec![] } else { vec![id - 1] },
             state,
-            step: step.map(str::to_owned),
-            build_log_id: None,
             started_at: None,
             finished_at: None,
             error: None,
+            approval_id: None,
+            inputs: vec![],
+            has_log: false,
         }
     }
 
     #[test]
-    fn render_dag_line_shows_chain_and_running_step() {
+    fn render_dag_line_shows_chain_with_derived_label_and_state() {
+        // Roll-up state (Running) + label ("rebuild") are derived from the node
+        // set — the wire no longer carries them. (`Done` nodes are included
+        // here to exercise glyph rendering; production filters them off.)
         let dag = DagView {
             id: 7,
-            kind: Template::Rebuild,
-            state: State::Running,
             source: Source::Manual,
             reason: "manual".to_owned(),
-            enqueued_at: 0,
-            started_at: Some(1),
+            created_at: from_secs(0),
+            started_at: Some(from_secs(1)),
             finished_at: None,
-            inputs: vec![],
-            approval_id: None,
             nodes: vec![
-                node(0, "alice", "prebuild", State::Done, None),
-                node(1, "alice", "stop_for_update", State::Done, None),
-                node(
-                    2,
-                    "alice",
-                    "swap",
-                    State::Running,
-                    Some("nixos-container update"),
-                ),
-                node(3, "alice", "reconcile", State::Queued, None),
+                node(0, "alice", "prebuild", State::Done),
+                node(1, "alice", "stop_for_update", State::Done),
+                node(2, "alice", "swap", State::Running),
+                node(3, "alice", "reconcile", State::Queued),
             ],
         };
         let line = render_dag_line(&dag);
         assert!(line.starts_with("▶ rebuild alice"), "{line}");
         assert!(
-            line.contains(
-                "✔ prebuild → ✔ stop_for_update → ▶ swap (nixos-container update) → ⏸ reconcile"
-            ),
+            line.contains("✔ prebuild → ✔ stop_for_update → ▶ swap → ⏸ reconcile"),
             "{line}"
         );
     }
 
     #[test]
     fn render_dag_line_surfaces_first_node_error() {
-        let mut failed = node(0, "bob", "prebuild", State::Failed, None);
+        let mut failed = node(0, "bob", "prebuild", State::Failed);
         failed.error = Some("nix build exploded".to_owned());
         let dag = DagView {
             id: 8,
-            kind: Template::Rebuild,
-            state: State::Failed,
             source: Source::Manual,
             reason: "manual".to_owned(),
-            enqueued_at: 0,
-            started_at: Some(1),
-            finished_at: Some(2),
-            inputs: vec![],
-            approval_id: None,
+            created_at: from_secs(0),
+            started_at: Some(from_secs(1)),
+            finished_at: Some(from_secs(2)),
             nodes: vec![failed],
         };
         let line = render_dag_line(&dag);