From 10b0f640af003ab08befe5a62591691a7bca2587 Mon Sep 17 00:00:00 2001 From: damocles Date: Mon, 3 Aug 2026 18:47:22 +0200 Subject: [PATCH] hivectl: migrate dag_progress to hive-jobq-wire's generic GraphNode --- Cargo.lock | 2 + hive-c0re/src/job_queue/mod.rs | 24 ++ hive-c0re/src/job_queue/model.rs | 9 + hive-c0re/src/server.rs | 1 + hive-host-sock/Cargo.toml | 1 + hive-host-sock/src/lib.rs | 28 +++ hivectl/Cargo.toml | 1 + hivectl/src/dag_progress.rs | 381 +++++++++++++++++-------------- 8 files changed, 272 insertions(+), 175 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 93b98034..22171566 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1725,6 +1725,7 @@ version = "0.1.0" dependencies = [ "chrono", "hive-jobq", + "hive-jobq-wire", "hive-sh4re", "hive-types", "serde", @@ -1861,6 +1862,7 @@ dependencies = [ "clap-markdown", "clap_complete", "hive-host-sock", + "hive-jobq-wire", "hive-sh4re", "hive-types", "http-body-util", diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index 6b2411df..e81a3bb6 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -378,6 +378,30 @@ impl JobQueue { .filter_map(|c| dag_view(&inner, c)) .collect() } + + /// One DAG's container node plus its live subtree, as generic wire + /// nodes — the `QueueNodes` polling surface behind `hivectl`'s + /// wait/progress loop. Sibling of [`Self::snapshot`] + /// (which serves the same graph through the typed `DagView`/`NodeView` + /// projection for the dashboard's `/api/state.rebuild_queue`), this one + /// goes through [`GraphWire::wire_snapshot`] instead — no `Done`-node + /// filtering, no roll-up field (the root's own `state` answers that, + /// see `hive_jobq_wire`'s doc comment). + /// + /// Empty when `dag_id` names no DAG container in the graph. Today that + /// only happens for a genuinely unknown id: nothing prunes the graph + /// yet (bounded-prune is a Stage-C follow-up, see [`visible_dags`]), so + /// a *completed* DAG's nodes keep riding here with a terminal `state` + /// rather than disappearing — callers watching for "done" should read + /// the root's `state`, not emptiness. + #[must_use] + pub fn dag_nodes(&self, dag_id: u64) -> Vec { + let inner = self.lock(); + let Some(root) = container(&inner, dag_id) else { + return Vec::new(); + }; + inner.graph().wire_snapshot([root]) + } } /// The container node of `dag_id` — the `NodeKind::Dag` root whose id equals diff --git a/hive-c0re/src/job_queue/model.rs b/hive-c0re/src/job_queue/model.rs index 9f45c3b1..a15e11d1 100644 --- a/hive-c0re/src/job_queue/model.rs +++ b/hive-c0re/src/job_queue/model.rs @@ -308,6 +308,15 @@ impl hive_jobq_wire::WireNode for NodeKind { { data.insert("inputs".to_owned(), inputs.clone().into()); } + // The DAG container's own metadata — nowhere else on the wire, since + // `GraphNode` carries no DAG-level fields (a group root is an + // ordinary node). `hivectl` needs `source` for its progress line; + // `reason` rides along for free rather than adding a second variant + // later for the one field the first pass missed. + if let NodeKind::Dag { source, reason, .. } = self { + data.insert("source".to_owned(), source.as_str().into()); + data.insert("reason".to_owned(), reason.clone().into()); + } // Not in the payload at all — the build log is keyed on node identity // in a side table, which is why `data` is handed the id. if let Some(log) = crate::build_logs::global().and_then(|h| h.id_for_node(id)) { diff --git a/hive-c0re/src/server.rs b/hive-c0re/src/server.rs index 26803aa2..d187d0b8 100644 --- a/hive-c0re/src/server.rs +++ b/hive-c0re/src/server.rs @@ -160,6 +160,7 @@ async fn dispatch(req: &HostRequest, coord: Arc) -> HostResponse { .collect(); HostResponse::dags(dags) } + HostRequest::QueueNodes { id } => HostResponse::nodes(coord.job_queue.dag_nodes(*id)), HostRequest::List => HostResponse::list(lifecycle::list().await?), // The agents root is ours and not world-traversable, so this // question is only answerable on this side of the socket — diff --git a/hive-host-sock/Cargo.toml b/hive-host-sock/Cargo.toml index ffd069fd..dd7fc555 100644 --- a/hive-host-sock/Cargo.toml +++ b/hive-host-sock/Cargo.toml @@ -10,6 +10,7 @@ workspace = true [dependencies] chrono.workspace = true hive-jobq.workspace = true +hive-jobq-wire.workspace = true hive-sh4re.workspace = true hive-types.workspace = true serde.workspace = true diff --git a/hive-host-sock/src/lib.rs b/hive-host-sock/src/lib.rs index 2363e3db..560029a7 100644 --- a/hive-host-sock/src/lib.rs +++ b/hive-host-sock/src/lib.rs @@ -214,6 +214,14 @@ pub enum HostRequest { /// `hivectl`'s wait/progress loop. A multi-step op is a single DAG /// (its whole graph in `nodes`). Result: [`HostResponse::dags`]. QueueDag { id: u64 }, + /// Fetch one job-queue DAG's container node plus its live subtree, as + /// generic `hive-jobq-wire` nodes — `hivectl`'s wait/progress loop. + /// Sibling of [`Self::QueueDag`]: same DAG, same + /// `id` (the container's own node id, what `queued_dags` already + /// carries), through the generic projection instead of the typed + /// `DagView`/`NodeView` (kept for `QueueDag`'s other consumer, + /// `/api/state.rebuild_queue`). Result: [`HostResponse::nodes`]. + QueueNodes { id: u64 }, /// List pending approval requests. Pending, /// Approve a pending request by id; the action runs immediately. @@ -539,6 +547,15 @@ pub struct HostResponse { /// been evicted from the queue's history tail. #[serde(default, skip_serializing_if = "Option::is_none")] pub dags: Option>, + /// `QueueNodes` result — the requested DAG's container node plus its + /// live subtree, as generic `hive-jobq-wire` nodes. `None` for every + /// other request kind. An empty `Vec` means `id` names no live DAG in + /// the graph (today: an unknown id — see `JobQueue::dag_nodes`'s doc + /// comment for why a *completed* DAG's nodes don't vanish the same way + /// `QueueDag`'s do); callers should read the root node's `state` for + /// terminality, not emptiness. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub nodes: Option>, /// Free-form operator-facing output lines the client prints verbatim /// (one per line). Carries results a request produced daemon-side that /// have no structured home — e.g. a freshly-minted matrix token, a @@ -639,6 +656,17 @@ impl HostResponse { } } + /// `QueueNodes` result — the polled DAG's container + subtree, as + /// generic wire nodes. + #[must_use] + pub fn nodes(nodes: Vec) -> Self { + Self { + ok: true, + nodes: Some(nodes), + ..Self::default() + } + } + /// A success carrying operator-facing output lines the client prints /// verbatim — the result shape for the `Matrix*` provisioning requests. #[must_use] diff --git a/hivectl/Cargo.toml b/hivectl/Cargo.toml index e282912d..baf32cf2 100644 --- a/hivectl/Cargo.toml +++ b/hivectl/Cargo.toml @@ -21,6 +21,7 @@ clap.workspace = true clap_complete.workspace = true clap-markdown = "0.1" hive-host-sock.workspace = true +hive-jobq-wire.workspace = true hive-sh4re.workspace = true hive-types.workspace = true http-body-util.workspace = true diff --git a/hivectl/src/dag_progress.rs b/hivectl/src/dag_progress.rs index aaca5710..fe4c3d4c 100644 --- a/hivectl/src/dag_progress.rs +++ b/hivectl/src/dag_progress.rs @@ -1,20 +1,30 @@ //! `hivectl` rebuild-queue progress rendering. //! //! Split out of `hivectl.rs` (which is already large): everything that -//! polls the daemon's DAG queue (`HostRequest::QueueDag`) and renders the +//! polls the daemon's DAG queue (`HostRequest::QueueNodes`) and renders the //! per-DAG / per-node progress lives here. [`wait_for_dags`] is the entry //! point the command handlers call; it dispatches to a live `indicatif` //! animation on a TTY and a plain line-on-change stream otherwise. +//! +//! Consumes `hive-jobq-wire`'s generic [`GraphNode`]/`NodePayload`: a +//! node's kind comes from `payload.label`, and node-specific extras +//! (`source`, `agent`) ride in `payload.data`'s opaque kvps — the shape +//! `hive-c0re`'s `impl WireNode for NodeKind` produces (see its doc +//! comment). There's no separate roll-up field on the wire: a node's own +//! `state` already reflects everything below it (see `hive_jobq_wire`'s +//! doc comment), so nothing here re-derives a roll-up or waits for every +//! node to go terminal before reading a result. use std::path::Path; use anyhow::{Context as _, Result, bail}; +use hive_host_sock::jobs::State; +use hive_jobq_wire::{GraphDep, GraphNode, WireId}; -/// Poll the submitted DAG ids (`HostRequest::QueueDag`, ~1s interval) and +/// Poll the submitted DAG ids (`HostRequest::QueueNodes`, ~1s interval) and /// render progress until they all reach a terminal state. Exits non-zero -/// (via the returned `Err`) when any DAG (or fan-out child) ends `failed`; -/// a `cancelled` DAG terminates the wait but is an operator action, not an -/// error. +/// (via the returned `Err`) when any DAG ends `failed`; a `cancelled` DAG +/// terminates the wait but is an operator action, not an error. /// /// # Errors /// @@ -34,6 +44,13 @@ pub(crate) async fn wait_for_dags(socket: &Path, ids: Vec, no_wait: bool) - } } +/// Find the DAG container among a `QueueNodes` response's nodes — the one +/// `GraphNode` with no structural parent. `wire_snapshot` hands back exactly +/// one per requested root, so this is a lookup, not a real search. +fn find_root(nodes: &[GraphNode]) -> Option<&GraphNode> { + nodes.iter().find(|n| n.parent.is_none()) +} + /// Non-TTY progress: print a fresh line whenever a DAG's rendered state /// changes. No cursor tricks, so it's clean in pipes and CI logs. async fn wait_for_dags_plain(socket: &Path, ids: Vec) -> Result<()> { @@ -42,38 +59,31 @@ async fn wait_for_dags_plain(socket: &Path, ids: Vec) -> Result<()> { let mut failed: Vec = Vec::new(); while !pending.is_empty() { for id in pending.clone() { - let resp = crate::client::request(socket, hive_host_sock::HostRequest::QueueDag { id }) - .await - .with_context(|| format!("connect to daemon socket {}", socket.display()))?; - let dags = resp.dags.unwrap_or_default(); - if dags.is_empty() { - // Evicted from the queue's history tail — it finished a - // while ago; nothing left to report on. + let resp = + crate::client::request(socket, hive_host_sock::HostRequest::QueueNodes { id }) + .await + .with_context(|| format!("connect to daemon socket {}", socket.display()))?; + let nodes = resp.nodes.unwrap_or_default(); + let Some(root) = find_root(&nodes) else { + // Unknown id — nothing prunes the graph yet (see + // `JobQueue::dag_nodes`'s doc comment), so an id that + // resolves to no container never named a real DAG. A + // *completed* DAG's nodes keep riding here instead, with a + // terminal root `state`, which is what the check below + // watches for. println!("job #{id}: gone from queue history"); pending.remove(&id); continue; + }; + let line = render_dag_line(root, &nodes); + if last.get(&id) != Some(&line) { + println!("{line}"); + last.insert(id, line); } - let mut all_terminal = true; - for d in &dags { - let line = render_dag_line(d); - if last.get(&d.id) != Some(&line) { - println!("{line}"); - last.insert(d.id, line); + if root.state.is_terminal() { + if root.state == State::Failed { + failed.push(format!("{} {}", dag_source(root), dag_agents(&nodes))); } - // Node-level terminality, not the roll-up: a DAG rolls - // up `failed` the moment one node fails while its - // after-any recovery node (rebuild's tail Reconcile) - // 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.rollup_state() == hive_host_sock::jobs::State::Failed { - failed.push(format!("{} {}", d.source.as_str(), dag_agents(d))); - } - } else { - all_terminal = false; - } - } - if all_terminal { pending.remove(&id); } } @@ -104,10 +114,9 @@ async fn wait_for_dags_animated(socket: &Path, ids: Vec) -> Result<()> { // insertion order groups a DAG's nodes right under its header. let mut dag_bars: std::collections::HashMap = std::collections::HashMap::new(); - let mut node_bars: std::collections::HashMap<(u64, hive_host_sock::jobs::NodeId), ProgressBar> = + let mut node_bars: std::collections::HashMap<(u64, WireId), ProgressBar> = std::collections::HashMap::new(); - let mut node_done: std::collections::HashSet<(u64, hive_host_sock::jobs::NodeId)> = - std::collections::HashSet::new(); + let mut node_done: std::collections::HashSet<(u64, WireId)> = std::collections::HashSet::new(); let mut pending: std::collections::BTreeSet = ids.into_iter().collect(); let mut failed: Vec = Vec::new(); @@ -115,62 +124,56 @@ async fn wait_for_dags_animated(socket: &Path, ids: Vec) -> Result<()> { while !pending.is_empty() { let now = now_unix(); for id in pending.clone() { - let resp = crate::client::request(socket, hive_host_sock::HostRequest::QueueDag { id }) - .await - .with_context(|| format!("connect to daemon socket {}", socket.display()))?; - let dags = resp.dags.unwrap_or_default(); - if dags.is_empty() { + let resp = + crate::client::request(socket, hive_host_sock::HostRequest::QueueNodes { id }) + .await + .with_context(|| format!("connect to daemon socket {}", socket.display()))?; + let nodes = resp.nodes.unwrap_or_default(); + let Some(root) = find_root(&nodes) else { mp.println(format!("job #{id}: gone from queue history")) .ok(); pending.remove(&id); continue; - } - let mut all_terminal = true; - for d in &dags { - let hdr = dag_bars.entry(d.id).or_insert_with(|| { + }; + let hdr = dag_bars.entry(id).or_insert_with(|| { + let b = mp.add(ProgressBar::new_spinner()); + b.set_style(plain.clone()); + b + }); + hdr.set_message(format!( + "{} {} {} · {}", + state_glyph(root.state), + dag_source(root), + dag_agents(&nodes), + fmt_dur(node_elapsed(root, now)), + )); + for n in nodes.iter().filter(|n| n.id != root.id) { + let key = (id, n.id); + if node_done.contains(&key) { + continue; + } + let bar = node_bars.entry(key).or_insert_with(|| { let b = mp.add(ProgressBar::new_spinner()); - b.set_style(plain.clone()); + b.set_style(spinner.clone()); + b.enable_steady_tick(std::time::Duration::from_millis(120)); b }); - hdr.set_message(format!( - "{} {} {} · {}", - state_glyph(d.rollup_state()), - d.source.as_str(), - dag_agents(d), - fmt_dur(dag_elapsed(d, now)), - )); - for n in &d.nodes { - let key = (d.id, n.id); - if node_done.contains(&key) { - continue; - } - let bar = node_bars.entry(key).or_insert_with(|| { - let b = mp.add(ProgressBar::new_spinner()); - b.set_style(spinner.clone()); - b.enable_steady_tick(std::time::Duration::from_millis(120)); - b - }); - if n.state.is_terminal() { - bar.set_style(plain.clone()); - bar.finish_with_message(format!( - " {} {}", - state_glyph(n.state), - node_line(d, n, now) - )); - node_done.insert(key); - } else { - bar.set_message(node_line(d, n, now)); - } - } - if d.nodes.iter().all(|n| n.state.is_terminal()) { - if d.rollup_state() == hive_host_sock::jobs::State::Failed { - failed.push(format!("{} {}", d.source.as_str(), dag_agents(d))); - } + if n.state.is_terminal() { + bar.set_style(plain.clone()); + bar.finish_with_message(format!( + " {} {}", + state_glyph(n.state), + node_line(&nodes, n, now) + )); + node_done.insert(key); } else { - all_terminal = false; + bar.set_message(node_line(&nodes, n, now)); } } - if all_terminal { + if root.state.is_terminal() { + if root.state == State::Failed { + failed.push(format!("{} {}", dag_source(root), dag_agents(&nodes))); + } pending.remove(&id); } } @@ -202,14 +205,34 @@ fn finish_wait(mut failed: Vec) -> Result<()> { } } -/// Distinct agents across a DAG's nodes, comma-joined for display — the -/// per-node replacement for the old DAG-level `agent` field. Single-agent -/// DAGs render one name; a hive-wide DAG lists each. -fn dag_agents(d: &hive_host_sock::jobs::DagView) -> String { +/// The DAG's `source` tag (`"manual"`, `"meta_update"`, …), read from the +/// container node's `payload.data` — see `hive-c0re`'s `impl WireNode for +/// NodeKind`'s `NodeKind::Dag` arm, the one place it's put on the wire. +/// Falls back to the label when absent (defensive; every real DAG +/// container sets it). +fn dag_source(root: &GraphNode) -> &str { + root.payload + .data + .get("source") + .and_then(serde_json::Value::as_str) + .unwrap_or(&root.payload.label) +} + +/// Distinct agents across a DAG's nodes, comma-joined for display. Each +/// node's `agent` (when it targets one) rides in `payload.data["agent"]` — +/// an opaque kvp, not a typed field, since `GraphNode` carries nothing +/// domain-specific (see `hive_jobq_wire::WireNode::data`'s doc comment). +fn dag_agents(nodes: &[GraphNode]) -> String { let mut seen: Vec<&str> = Vec::new(); - for n in &d.nodes { - if !seen.contains(&n.agent.as_str()) { - seen.push(&n.agent); + for n in nodes { + if let Some(agent) = n + .payload + .data + .get("agent") + .and_then(serde_json::Value::as_str) + && !seen.contains(&agent) + { + seen.push(agent); } } seen.join(",") @@ -224,23 +247,13 @@ fn now_unix() -> i64 { .unwrap_or(0) } -/// Elapsed seconds for a DAG: `started_at` (falling back to `created_at`) -/// through `finished_at` or `now`. The wire carries these as RFC3339 -/// `DateTime`; compare in unix seconds against `now`. -fn dag_elapsed(d: &hive_host_sock::jobs::DagView, now: i64) -> i64 { - 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_host_sock::jobs::NodeView, now: i64) -> i64 { - match n.started_at { - Some(start) => (n.finished_at.map_or(now, |t| t.timestamp()) - start.timestamp()).max(0), - None => 0, - } +/// Elapsed seconds for a node: `started_at` → `finished_at`/`now` once +/// it's run; `created_at` → `now` while it's still queued. `GraphNode` +/// carries no group-level timestamp separate from its own, so a header +/// line just reads this off whichever node it's summarizing. +fn node_elapsed(n: &GraphNode, now: i64) -> i64 { + let start = n.started_at.unwrap_or(n.created_at); + (n.finished_at.map_or(now, |t| t.timestamp()) - start.timestamp()).max(0) } /// Compact duration: `45s` under a minute, else `1m03s`. @@ -254,20 +267,26 @@ fn fmt_dur(secs: i64) -> String { } /// 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_host_sock::jobs::DagView, - n: &hive_host_sock::jobs::NodeView, - now: i64, -) -> String { +/// (>1 node-dependency), its elapsed timer, and a truncated error tail. +/// Resource deps (`GraphDep::Resource`) don't name another node, so they're +/// filtered out of the fan-in count the same way `hive-c0re`'s own +/// `dag_view` projection already does. +fn node_line(nodes: &[GraphNode], n: &GraphNode, now: i64) -> String { use std::fmt::Write as _; - let mut s = n.kind.clone(); - if n.deps.len() > 1 { - let after: Vec<&str> = n - .deps + let mut s = n.payload.label.clone(); + let dep_ids: Vec = n + .deps + .iter() + .filter_map(|d| match d { + GraphDep::Node { id, .. } => Some(*id), + GraphDep::Resource { .. } => None, + }) + .collect(); + if dep_ids.len() > 1 { + let after: Vec<&str> = dep_ids .iter() - .filter_map(|dep| d.nodes.iter().find(|m| m.id == *dep)) - .map(|m| m.kind.as_str()) + .filter_map(|id| nodes.iter().find(|m| m.id == *id)) + .map(|m| m.payload.label.as_str()) .collect(); if !after.is_empty() { let _ = write!(s, " (after {})", after.join(", ")); @@ -284,38 +303,46 @@ fn node_line( s } -fn state_glyph(state: hive_host_sock::jobs::State) -> &'static str { +fn state_glyph(state: State) -> &'static str { match state { - hive_host_sock::jobs::State::Pending => "⏸", + State::Pending => "⏸", // `Finishing` is own-work-done with sub-nodes still going — in flight, // so it reads the same as running. - hive_host_sock::jobs::State::Running | hive_host_sock::jobs::State::Finishing => "▶", - hive_host_sock::jobs::State::Done => "✔", - hive_host_sock::jobs::State::Failed => "✖", - hive_host_sock::jobs::State::Cancelled => "⊘", + State::Running | State::Finishing => "▶", + State::Done => "✔", + State::Failed => "✖", + State::Cancelled => "⊘", // Distinct from cancelled: nothing went wrong, this branch just // wasn't the one the run took. - hive_host_sock::jobs::State::Skipped => "·", + State::Skipped => "·", } } /// One progress line for a DAG: roll-up glyph, `source`, agents, then the -/// node chain — the CLI twin of the dashboard's queue card. The header shows -/// what the backend sends (`source` + the raw node kinds); only the roll-up -/// state glyph is derived from the node set. Used by the plain (non-TTY) path. -fn render_dag_line(d: &hive_host_sock::jobs::DagView) -> String { +/// node chain — the CLI twin of the dashboard's queue card. The glyph and +/// `source` come off `root` (the one entry in `nodes` with no parent); +/// everything else comes straight off the wire. Used by the plain +/// (non-TTY) path. +/// +/// The chain shows every entry `wire_snapshot` sends, `Done` ones +/// included — that filtering was a dashboard-history-bounding concern, +/// not relevant to a single actively-watched job, so a completed step +/// keeps its checkmark instead of vanishing from the line, matching how +/// the animated path already behaves. +fn render_dag_line(root: &GraphNode, nodes: &[GraphNode]) -> String { use std::fmt::Write as _; let mut out = format!( "{} {} {:<12}", - state_glyph(d.rollup_state()), - d.source.as_str(), - dag_agents(d) + state_glyph(root.state), + dag_source(root), + dag_agents(nodes) ); - for (i, n) in d.nodes.iter().enumerate() { + let children: Vec<&GraphNode> = nodes.iter().filter(|n| n.id != root.id).collect(); + for (i, n) in children.iter().enumerate() { let sep = if i == 0 { " " } else { " → " }; - let _ = write!(out, "{sep}{} {}", state_glyph(n.state), n.kind); + let _ = write!(out, "{sep}{} {}", state_glyph(n.state), n.payload.label); } - if let Some(err) = d.nodes.iter().find_map(|n| n.error.as_deref()) { + if let Some(err) = children.iter().find_map(|n| n.error.as_deref()) { let short: String = err.chars().take(120).collect(); let _ = write!(out, " — {short}"); } @@ -324,50 +351,61 @@ fn render_dag_line(d: &hive_host_sock::jobs::DagView) -> String { #[cfg(test)] mod tests { - use hive_host_sock::jobs::{DagView, NodeView, Source, State}; - use hive_sh4re::wire_time::from_secs; + use hive_host_sock::jobs::State; + use hive_jobq_wire::{GraphNode, NodePayload}; + use serde_json::json; use super::render_dag_line; - fn node(id: u64, agent: &str, kind: &str, state: State) -> NodeView { - NodeView { + fn dag_root(id: u64, source: &str, state: State) -> GraphNode { + GraphNode { id, parent: None, - agent: agent.to_owned(), - kind: kind.to_owned(), - deps: if id == 0 { vec![] } else { vec![id - 1] }, state, + deps: Vec::new(), + created_at: hive_sh4re::wire_time::from_secs(0), started_at: None, finished_at: None, error: None, - approval_id: None, - inputs: vec![], - build_log_id: None, + payload: NodePayload { + label: "dag".to_owned(), + data: json!({ "source": source }), + }, + } + } + + fn work_node(id: u64, root: u64, agent: &str, label: &str, state: State) -> GraphNode { + GraphNode { + id, + parent: Some(root), + state, + deps: Vec::new(), + created_at: hive_sh4re::wire_time::from_secs(0), + started_at: None, + finished_at: None, + error: None, + payload: NodePayload { + label: label.to_owned(), + data: json!({ "agent": agent }), + }, } } #[test] fn render_dag_line_shows_source_and_chain() { - // The header shows what the backend sends: the roll-up state glyph - // (derived — Running here) + the DAG `source` ("manual"); the operation - // is read off the node chain, not a client-side label. (`Done` nodes - // are included here to exercise glyph rendering; production filters - // them off.) - let dag = DagView { - id: 7, - source: Source::Manual, - reason: "manual".to_owned(), - created_at: from_secs(0), - started_at: Some(from_secs(1)), - finished_at: None, - nodes: vec![ - 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::Pending), - ], - }; - let line = render_dag_line(&dag); + // The header shows what the backend sends: the glyph off `root`'s + // own `state` (`Running` here) + `source` ("manual"); the operation + // is read off the node chain, not a client-side label. `Done` + // nodes stay in the chain — nothing here filters them off. + let root = dag_root(7, "manual", State::Running); + let nodes = vec![ + root.clone(), + work_node(0, 7, "alice", "prebuild", State::Done), + work_node(1, 7, "alice", "stop_for_update", State::Done), + work_node(2, 7, "alice", "swap", State::Running), + work_node(3, 7, "alice", "reconcile", State::Pending), + ]; + let line = render_dag_line(&root, &nodes); assert!(line.starts_with("▶ manual alice"), "{line}"); assert!( line.contains("✔ prebuild → ✔ stop_for_update → ▶ swap → ⏸ reconcile"), @@ -377,18 +415,11 @@ mod tests { #[test] fn render_dag_line_surfaces_first_node_error() { - let mut failed = node(0, "bob", "prebuild", State::Failed); + let root = dag_root(8, "manual", State::Failed); + let mut failed = work_node(0, 8, "bob", "prebuild", State::Failed); failed.error = Some("nix build exploded".to_owned()); - let dag = DagView { - id: 8, - source: Source::Manual, - reason: "manual".to_owned(), - 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); + let nodes = vec![root.clone(), failed]; + let line = render_dag_line(&root, &nodes); assert!(line.contains("✖ manual"), "{line}"); assert!(line.contains("— nix build exploded"), "{line}"); }